package p.ithorns.integration.hazelcast.config;

import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.jet.Job;
import com.hazelcast.jet.cdc.CdcSinks;
import com.hazelcast.jet.cdc.ChangeRecord;
import com.hazelcast.jet.cdc.mysql.MySqlCdcSources;
import com.hazelcast.jet.config.JobConfig;
import com.hazelcast.jet.pipeline.Pipeline;
import com.hazelcast.jet.pipeline.StreamSource;
import org.springframework.context.annotation.Bean;
import p.ithorns.integration.db.entity.User;
import p.ithorns.integration.db.model.Customer;

/**
 * CdcConfig
 *
 * @author Ht.L
 * @date 2023-10-17 20:58
 * @since 1.0.0
 */
//@Configuration
public class CdcConfig {

    /**
     * 需要在配置文件开启Jet
     * hazelcast.jet.enable=true
     */
    private StreamSource<ChangeRecord> mysqlSource(String name, String... tbs) {
        return MySqlCdcSources.mysql(name)
                .setDatabaseAddress("127.0.0.1")
                .setDatabasePort(3306)
                .setDatabaseUser("root")
                .setDatabasePassword("Z@9e59b$")
                .setClusterName("local")
                .setDatabaseWhitelist("hazelcast")
                .setTableWhitelist(tbs)
                .setCustomProperty("database.connectionTimeZone", "Asia/Shanghai")
                .setCustomProperty("database.allowPublicKeyRetrieval", "true")
                .build();
    }

    //@Bean
    public Job customers(HazelcastInstance hz) {
        Pipeline pipeline = Pipeline.create();
        StreamSource<ChangeRecord> source = mysqlSource("s1", "hazelcast.users");
        pipeline.readFrom(source)
                .withoutTimestamps()
                .peek()
                .writeTo(CdcSinks.map("customers",
                        r -> r.key().toMap().get("id"),
                        r -> r.value().toObject(Customer.class).toString()));
        JobConfig cfg = new JobConfig().setName("mysql-customers");
        // HazelcastInstance hz = Hazelcast.bootstrappedInstance();
        return hz.getJet().newJob(pipeline, cfg);
    }

    @Bean
    public Job users(HazelcastInstance hz) {
        StreamSource<ChangeRecord> source = mysqlSource("s2", "hazelcast.customers");
        Pipeline pipeline = Pipeline.create();
        pipeline.readFrom(source)
                .withoutTimestamps()
                .peek()
                .writeTo(CdcSinks.map("users",
                        r -> r.key().toMap().get("id"),
                        r -> r.value().toObject(User.class).toString()));
        JobConfig cfg = new JobConfig().setName("mysql-users");
        // HazelcastInstance hz = Hazelcast.bootstrappedInstance();
        return hz.getJet().newJob(pipeline, cfg);
    }

}